// Copyright (c) 2023 Institute of Software, Chinese Academy of Sciences
// Ripple is licensed under Mulan PSL v2.
// You can use this software according to the terms and conditions of the Mulan PSL v2.
// You may obtain a copy of Mulan PSL v2 at:
//          http://license.coscl.org.cn/MulanPSL2
// THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
// EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
// MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
// See the Mulan PSL v2 for more details.
package ripple.server.helper;

import java.net.InetSocketAddress;
import java.text.SimpleDateFormat;
import java.util.UUID;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import io.netty.channel.Channel;
import io.netty.channel.socket.nio.NioSocketChannel;
import ripple.common.entity.AbstractMessage;
import ripple.common.entity.Constants;
import ripple.common.entity.DataMessage;
import ripple.common.entity.IncrementalUpdateMessage;
import ripple.common.entity.ReformRequest;
import ripple.common.entity.UpdateMessage;
import ripple.common.tcp.message.DispatchRequest;
import ripple.common.tcp.message.HeartbeatRequest;
import ripple.common.tcp.message.MQSyncRequest;
import ripple.common.tcp.message.ReformSyncRequest;
import ripple.common.tcp.message.SyncRequest;
import ripple.server.tcp.message.AckRequest;

/**
 * @author Zhen Tang
 */
public class Api {

    private static final Logger LOGGER = LoggerFactory.getLogger(Api.class);

    private Api() {

    }

    public static void heartbeat(Channel channel) {
        HeartbeatRequest heartbeatRequest = new HeartbeatRequest();
        heartbeatRequest.setUuid(UUID.randomUUID());
//        InetSocketAddress localAddress = ((NioSocketChannel) channel).localAddress();
//        InetSocketAddress remoteAddress = ((NioSocketChannel) channel).remoteAddress();
//        LOGGER.info("[Api] [{}:{}<-->{}:{}] Send HEARTBEAT request. UUID = {}."
//                , localAddress.getHostString(), localAddress.getPort(), remoteAddress.getHostString()
//                , remoteAddress.getPort(), heartbeatRequest.getUuid());
        channel.writeAndFlush(heartbeatRequest);
    }

    public static void sync(Channel channel, AbstractMessage message) {
        SyncRequest syncRequest = new SyncRequest();
        syncRequest.setUuid(UUID.randomUUID());
        syncRequest.setMessageUuid(message.getUuid());
        syncRequest.setOperationType(message.getType());
        syncRequest.setApplicationName(message.getApplicationName());
        syncRequest.setKey(message.getKey());
        if (message instanceof UpdateMessage) {
            syncRequest.setValue(((UpdateMessage) message).getValue());

            syncRequest.setFromWhere(message.getFromWhere());
            // DebugTool.debugInfo("[Api] sync message from " + message.getFromWhere());

        } else if (message instanceof IncrementalUpdateMessage) {
            syncRequest.setBaseMessageUuid(((IncrementalUpdateMessage) message).getBaseMessageUuid());
            syncRequest.setAtomicOperation(((IncrementalUpdateMessage) message).getAtomicOperation());
            syncRequest.setValue(((IncrementalUpdateMessage) message).getValue());
        }
        syncRequest.setLastUpdate(message.getLastUpdate());
        syncRequest.setLastUpdateServerId(message.getLastUpdateServerId());

//        InetSocketAddress localAddress = ((NioSocketChannel) channel).localAddress();
//        InetSocketAddress remoteAddress = ((NioSocketChannel) channel).remoteAddress();
//        LOGGER.info("[Api] [{}:{}<-->{}:{}] Send SYNC request. UUID = {}, Message UUID = {}" +
//                        ", Operation Type = {}, Application Name = {}, Key = {}, Base Message UUID = {}" +
//                        ", Atomic Operation = {}, Value = {}, Last Update = {}" +
//                        ", Last Update Server Id = {}."
//                , localAddress.getHostString(), localAddress.getPort(), remoteAddress.getHostString()
//                , remoteAddress.getPort(), syncRequest.getUuid(), syncRequest.getMessageUuid()
//                , syncRequest.getOperationType(), syncRequest.getApplicationName(), syncRequest.getKey()
//                , syncRequest.getBaseMessageUuid(), syncRequest.getAtomicOperation()
//                , syncRequest.getValue(), SimpleDateFormat.getDateTimeInstance().format(syncRequest.getLastUpdate())
//                , syncRequest.getLastUpdateServerId());
        channel.writeAndFlush(syncRequest);
    }

    public static void MQSync(Channel channel, AbstractMessage message) {
        MQSyncRequest mqSyncRequest = new MQSyncRequest();
        mqSyncRequest.setUuid(UUID.randomUUID());
        mqSyncRequest.setMessageUuid(message.getUuid());
        mqSyncRequest.setOperationType(message.getType());
        mqSyncRequest.setTopic(message.getTopic());
        if (message instanceof DataMessage) {
            mqSyncRequest.setData(((DataMessage) message).getData());
            mqSyncRequest.setTowards(((DataMessage) message).getTowards());
//        } else if (message instanceof IncrementalUpdateMessage) {
//            MQSyncRequest.setBaseMessageUuid(((IncrementalUpdateMessage) message).getBaseMessageUuid());
//            MQSyncRequest.setAtomicOperation(((IncrementalUpdateMessage) message).getAtomicOperation());
//            MQSyncRequest.setValue(((IncrementalUpdateMessage) message).getValue());
        }
        mqSyncRequest.setLastUpdate(message.getLastUpdate());
        mqSyncRequest.setLastUpdateServerId(message.getLastUpdateServerId());
        InetSocketAddress localAddress = ((NioSocketChannel) channel).localAddress();
        InetSocketAddress remoteAddress = ((NioSocketChannel) channel).remoteAddress();
        LOGGER.info("[Api] [{}:{}<-->{}:{}] Send MQ SYNC request. UUID = {}, Message UUID = {}"
                + ", Operation Type = {}, Topic = {}, Data String = {}, Base Message UUID = {}"
                + ", Atomic Operation = {}, Last Update = {}"
                + ", Last Update Server Id = {}.",
                localAddress.getHostString(), localAddress.getPort(), remoteAddress.getHostString(),
                remoteAddress.getPort(), mqSyncRequest.getUuid(), mqSyncRequest.getMessageUuid(),
                mqSyncRequest.getOperationType(), mqSyncRequest.getData(), mqSyncRequest.getData().toString(),
                mqSyncRequest.getBaseMessageUuid(), mqSyncRequest.getAtomicOperation(),
                SimpleDateFormat.getDateTimeInstance().format(mqSyncRequest.getLastUpdate()),
                mqSyncRequest.getLastUpdateServerId());
        channel.writeAndFlush(mqSyncRequest);
    }

    public static void reformSync(Channel channel, AbstractMessage message) {
        ReformSyncRequest reformSyncRequest = new ReformSyncRequest();
        reformSyncRequest.setUuid(UUID.randomUUID());
        reformSyncRequest.setMessageUuid(message.getUuid());
        reformSyncRequest.setOperationType(message.getType());
//        if (message.getType().equals(Constants.REFORM_PREPARE)) {
//        } else if (message.getType().equals(Constants.REFORM_PROPOSE)) {
//        } else if (message.getType().equals(Constants.REFORM_COMMIT)) {
//        } else if (message.getType().equals(Constants.REFORM_DEPRECATE)) {
//        }
        if (message.getType().equals(Constants.REFORM_PROPOSE)) {
            reformSyncRequest.setNodeList(((ReformRequest) message).getNodeList());
        }
        reformSyncRequest.setLastUpdate(message.getLastUpdate());
        reformSyncRequest.setLastUpdateServerId(message.getLastUpdateServerId());
//        InetSocketAddress localAddress = ((NioSocketChannel) channel).localAddress();
//        InetSocketAddress remoteAddress = ((NioSocketChannel) channel).remoteAddress();
//        LOGGER.info("[Api] [{}:{}<-->{}:{}] Send MQ SYNC request. UUID = {}, Message UUID = {}" +
//                        ", Operation Type = {}, Topic = {}, Data String = {}, Base Message UUID = {}" +
//                        ", Atomic Operation = {}, Last Update = {}" +
//                        ", Last Update Server Id = {}."
//                , localAddress.getHostString(), localAddress.getPort(), remoteAddress.getHostString()
//                , remoteAddress.getPort(), mqSyncRequest.getUuid(), mqSyncRequest.getMessageUuid()
//                , mqSyncRequest.getOperationType(), mqSyncRequest.getData(), mqSyncRequest.getData().toString()
//                , mqSyncRequest.getBaseMessageUuid(), mqSyncRequest.getAtomicOperation()
//                , SimpleDateFormat.getDateTimeInstance().format(mqSyncRequest.getLastUpdate())
//                , mqSyncRequest.getLastUpdateServerId());
        channel.writeAndFlush(reformSyncRequest);
    }

    public static void ack(Channel channel, UUID messageUuid, int sourceId, int nodeId) {
        AckRequest ackRequest = new AckRequest();
        ackRequest.setUuid(UUID.randomUUID());
        ackRequest.setMessageUuid(messageUuid);
        ackRequest.setSourceId(sourceId);
        ackRequest.setNodeId(nodeId);
//        InetSocketAddress localAddress = ((NioSocketChannel) channel).localAddress();
//        InetSocketAddress remoteAddress = ((NioSocketChannel) channel).remoteAddress();
//        LOGGER.info("[Api] [{}:{}<-->{}:{}] Send ACK request. UUID = {}, Message UUID = {}" +
//                        ", Source ID = {}, Node ID = {}."
//                , localAddress.getHostString(), localAddress.getPort(), remoteAddress.getHostString()
//                , remoteAddress.getPort(), ackRequest.getUuid(), ackRequest.getMessageUuid()
//                , ackRequest.getSourceId(), ackRequest.getNodeId());
        channel.writeAndFlush(ackRequest);
    }

    public static void dispatch(Channel channel, String clientListSignature, AbstractMessage message) {
        DispatchRequest dispatchRequest = new DispatchRequest();
        dispatchRequest.setUuid(UUID.randomUUID());
        dispatchRequest.setClientListSignature(clientListSignature);
        dispatchRequest.setMessageUuid(message.getUuid());
        dispatchRequest.setOperationType(message.getType());
        dispatchRequest.setApplicationName(message.getApplicationName());
        dispatchRequest.setKey(message.getKey());
        if (message instanceof UpdateMessage) {
            dispatchRequest.setValue(((UpdateMessage) message).getValue());
        } else if (message instanceof IncrementalUpdateMessage) {
            dispatchRequest.setBaseMessageUuid(((IncrementalUpdateMessage) message).getBaseMessageUuid());
            dispatchRequest.setAtomicOperation(((IncrementalUpdateMessage) message).getAtomicOperation());
            dispatchRequest.setValue(((IncrementalUpdateMessage) message).getValue());
        }
        dispatchRequest.setLastUpdate(message.getLastUpdate());
        dispatchRequest.setLastUpdateServerId(message.getLastUpdateServerId());
//        InetSocketAddress localAddress = ((NioSocketChannel) channel).localAddress();
//        InetSocketAddress remoteAddress = ((NioSocketChannel) channel).remoteAddress();
//        LOGGER.info("[Api] [{}:{}<-->{}:{}] Send DISPATCH request. UUID = {}, Client List Signature = {}, Message UUID = {}" +
//                        ", Operation Type = {}, Application Name = {}, Key = {}, Base Message UUID = {}" +
//                        ", Atomic Operation = {}, Value = {}, Last Update = {}" +
//                        ", Last Update Server Id = {}."
//                , localAddress.getHostString(), localAddress.getPort(), remoteAddress.getHostString()
//                , remoteAddress.getPort(), dispatchRequest.getUuid(), dispatchRequest.getClientListSignature(), dispatchRequest.getMessageUuid()
//                , dispatchRequest.getOperationType(), dispatchRequest.getApplicationName(), dispatchRequest.getKey()
//                , dispatchRequest.getBaseMessageUuid(), dispatchRequest.getAtomicOperation()
//                , dispatchRequest.getValue(), SimpleDateFormat.getDateTimeInstance().format(dispatchRequest.getLastUpdate())
//                , dispatchRequest.getLastUpdateServerId());
        channel.writeAndFlush(dispatchRequest);
    }
}
